衣舞晨风
0%

Kubernetes中Java应用Heap Dump

发表于 更新于 分类于
本文字数： 8.7k 阅读时长 ≈ 8 分钟

Dump Java Heap to OSS

伴随着微服务及容器化的发展，越来越多的应用运行在kubernetes集群中，运维、调试的问题也随之而来。以Java为例，当线上环境出现内存问题，比如OOM，这时候需要Dump内存进行分析的时候，就会发现对于普通开发人员来说他们没有操作kubernetes集群机器的权限，从而导致，Dump出来的文件无法回传到开发手中进行MAT之类的分析。

本文的解决办法是这样的，当用户需要Dump某个应用实例的时候，只需要在实例终端界面点击一下按钮，后台会自动Dump Heap到OSS上，上传完成后，会将下载的信息展示在列表页，这时候开发人员就可以进行下载了。

具体的操作流程是这样的：
1、检测Pod中是否有JDK TOOLS
2、拷贝Dump工具到对应的Pod中
3、赋予Dump工具可执行权限
4、运行Dump工具

Dump工具会识别Java进程，如果有多个Java进程会Dump进程号小的那一个。

核心代码主要是拷贝Dump工具到对应的Pod中

jmap.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package dump

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"strings"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/remotecommand"
)

var jmapDumpTool = "jmap-dump-tool"

func init() {
	// jmap-dump-tool 文件名
	if os.Getenv("JMAP_DUMP_TOOL_NAME") != "" {
		jmapDumpTool = os.Getenv("JMAP_DUMP_TOOL_NAME")
	}
	log.Println("JMAP_DUMP_TOOL_NAME:" + jmapDumpTool)
}

func DumpJavaHeap(ctx context.Context, project, app, env, cluster, namespace, podID, container string) error {
   
	// todo 获取k8s信息部分 需要替换成自己的
	ctxName := meta.GetContextName(cluster)
	kclient, err := k8s.GetClient(ctxName)
	if err != nil {
		log.Println(err)
		return errors.New("获取kubernetes client失败！")
	}

	// 获取kube config配置
	config, err := k8s.GetClientConfig(ctxName)
	if err != nil {
		log.Println(err)
		return errors.New("获取kube config失败！")
	}

	pod := new(pod)
	pod.Namespace = namespace
	pod.Name = podID
	pod.ContainerName = container

	log.Println("开始检测目标容器中是否有JDK Tool")
	//检测是否 容器中是否有jps命令
	cmds := []string{"sh", "-c", "jps"}
	req := kclient.CoreV1().RESTClient().
		Post().
		Namespace(pod.Namespace).
		Resource("pods").
		Name(pod.Name).
		SubResource("exec").
		VersionedParams(&corev1.PodExecOptions{
			Container: pod.ContainerName,
			Command:   cmds,
			Stdin:     true,
			Stdout:    true,
			Stderr:    true,
			TTY:       false,
		}, scheme.ParameterCodec)

	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
	if err != nil {
		log.Println(err)
		return &customerr.JavaHeapDumpError{Msg: "检查对应容器中是否有JDK Tools时发生异常！"}
	}
	bufErr := new(bytes.Buffer)
	err = exec.Stream(remotecommand.StreamOptions{
		Stdin:  strings.NewReader(""),
		Stdout: os.Stdout,
		Stderr: bufErr,
		Tty:    false,
	})
	if bufErr.Len() > 0 {
		e := string(bufErr.Bytes())
		if e != "" {
			log.Println(e)
			return &customerr.JavaHeapDumpError{Msg: "请检查对应容器中是否有JDK Tools，" + e}
		}
	}

	log.Println("检测完成，目标容器中有JDK Tool")

	srcPath := "/" + jmapDumpTool
	log.Println("srcPath:" + srcPath)
	destPath := "/" + jmapDumpTool
	log.Println("destPath:" + destPath)

	err = pod.copyToPod(ctx, kclient, config, srcPath, destPath)
	if err != nil {
		log.Println(err)
		return &customerr.JavaHeapDumpError{Msg: "dump工具拷贝失败！"}
	}
	log.Println("jmap-dump-tool copy to pod end")
	// 赋予可执行权限
	cmds = []string{"sh", "-c", "chmod +x /" + jmapDumpTool}
	err = pod.Exec(ctx, kclient, config, cmds)
	if err != nil {
		log.Println(err)
		return &customerr.JavaHeapDumpError{Msg: "dump工具赋予可执行权限失败！"}
	}
	log.Println("jmap-dump-tool 已赋予可执行权限")
	go execDump(*pod, jmapDumpTool, project, app, env, kclient, config)
	return nil
}

func execDump(p pod, jmapDumpTool, project, app, env string, client *kubernetes.Clientset, config *rest.Config) {
	// 执行
	c := fmt.Sprintf("/%s %s %s %s", jmapDumpTool, project, app, env)
	cmds := []string{"sh", "-c", c}
	err := p.Exec(context.Background(), client, config, cmds)
	if err != nil {
		log.Println(err)
	}
}

cp.go（参考kubectl cp的实现）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
package dump

import (
	"archive/tar"
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"os"
	"path"
	"strings"

	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/remotecommand"
)

type pod struct {
	Name          string
	Namespace     string
	ContainerName string
}

func (i *pod) copyToPod(ctx context.Context, client *kubernetes.Clientset, config *rest.Config, srcPath string, destPath string) error {
	reader, writer := io.Pipe()

	if destPath != "/" && strings.HasSuffix(string(destPath[len(destPath)-1]), "/") {
		destPath = destPath[:len(destPath)-1]
	}

	if err := checkDestinationIsDir(ctx, client, config, i, destPath); err == nil {
		destPath = destPath + "/" + path.Base(srcPath)
	}

	go func() {
		defer writer.Close()
		err := makeTar(srcPath, destPath, writer)
		if err != nil {
			fmt.Println(err)
		}
	}()

	var cmdArr []string

	cmdArr = []string{"tar", "-xf", "-"}
	destDir := path.Dir(destPath)
	if len(destDir) > 0 {
		cmdArr = append(cmdArr, "-C", destDir)
	}
	//remote shell.
	req := client.CoreV1().RESTClient().
		Post().
		Namespace(i.Namespace).
		Resource("pods").
		Name(i.Name).
		SubResource("exec").
		VersionedParams(&corev1.PodExecOptions{
			Container: i.ContainerName,
			Command:   cmdArr,
			Stdin:     true,
			Stdout:    true,
			Stderr:    true,
			TTY:       false,
		}, scheme.ParameterCodec)

	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
	if err != nil {
		return err
	}

	err = exec.Stream(remotecommand.StreamOptions{
		Stdin:  reader,
		Stdout: os.Stdout,
		Stderr: os.Stderr,
		Tty:    false,
	})
	if err != nil {
		return err
	}
	return nil
}

func checkDestinationIsDir(ctx context.Context, client *kubernetes.Clientset, config *rest.Config, i *pod, destPath string) error {
	return i.Exec(ctx, client, config, []string{"test", "-d", destPath})
}

func makeTar(srcPath, destPath string, writer io.Writer) error {
	// TODO: use compression here?
	tarWriter := tar.NewWriter(writer)
	defer tarWriter.Close()

	srcPath = path.Clean(srcPath)
	destPath = path.Clean(destPath)
	return recursiveTar(path.Dir(srcPath), path.Base(srcPath), path.Dir(destPath), path.Base(destPath), tarWriter)
}

func recursiveTar(srcBase, srcFile, destBase, destFile string, tarWriter *tar.Writer) error {

	filepath := path.Join(srcBase, srcFile)
	stat, err := os.Lstat(filepath)
	if err != nil {
		return err
	}
	if stat.IsDir() {
		files, err := ioutil.ReadDir(filepath)
		if err != nil {
			return err
		}
		if len(files) == 0 {
			//case empty directory
			hdr, _ := tar.FileInfoHeader(stat, filepath)
			hdr.Name = destFile
			if err := tarWriter.WriteHeader(hdr); err != nil {
				return err
			}
		}
		for _, f := range files {
			if err := recursiveTar(srcBase, path.Join(srcFile, f.Name()), destBase, path.Join(destFile, f.Name()), tarWriter); err != nil {
				return err
			}
		}
		return nil
	} else if stat.Mode()&os.ModeSymlink != 0 {
		//case soft link
		hdr, _ := tar.FileInfoHeader(stat, filepath)
		target, err := os.Readlink(filepath)
		if err != nil {
			return err
		}

		hdr.Linkname = target
		hdr.Name = destFile
		if err := tarWriter.WriteHeader(hdr); err != nil {
			return err
		}
	} else {
		//case regular file or other file type like pipe
		hdr, err := tar.FileInfoHeader(stat, filepath)
		if err != nil {
			return err
		}
		hdr.Name = destFile
		err = tarWriter.WriteHeader(hdr)
		if err != nil {
			log.Println(err)
			return err
		}

		f, err := os.Open(filepath)
		if err != nil {
			return err
		}
		defer f.Close()

		if _, err := io.Copy(tarWriter, f); err != nil {
			return err
		}
		return f.Close()
	}
	return nil
}

func (i *pod) Exec(ctx context.Context, client *kubernetes.Clientset, config *rest.Config, cmd []string) error {

	req := client.CoreV1().RESTClient().
		Post().
		Namespace(i.Namespace).
		Resource("pods").
		Name(i.Name).
		SubResource("exec").
		VersionedParams(&corev1.PodExecOptions{
			Container: i.ContainerName,
			Command:   cmd,
			Stdin:     true,
			Stdout:    true,
			Stderr:    true,
			TTY:       false,
		}, scheme.ParameterCodec)

	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
	if err != nil {
		return err
	}

	err = exec.Stream(remotecommand.StreamOptions{
		Stdin:  strings.NewReader(""),
		Stdout: os.Stdout,
		Stderr: os.Stderr,
		Tty:    false,
	})

	if err != nil {
		return err
	}
	return nil
}
相关文章

欢迎关注我的其它发布渠道